package com.gjy.design.observer.disruptor;

import com.lmax.disruptor.EventHandler;

import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-11-08 09:15:57
 */
public class SubscriberEventHandler implements EventHandler<DisruptorEvent> {

    private final String handlerName;
    private final AtomicBoolean running = new AtomicBoolean(true);

    public SubscriberEventHandler(String handlerName) {
        this.handlerName = handlerName;
    }

    @Override
    public void onEvent(DisruptorEvent event, long sequence, boolean endOfBatch) {
        if (!running.get()) return;

        MessageEvent messageEvent = event.getMessageEvent();
        CopyOnWriteArraySet<Subscriber> subscribers = event.getSubscribers();

        if (messageEvent != null && subscribers != null) {
            for (Subscriber subscriber : subscribers) {
                try {
                    subscriber.onMessage(messageEvent);
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        }
    }

    public void shutdown() {
        running.set(false);
    }

}
